{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "ename": "AttributeError",
     "evalue": "module 'asyncio' has no attribute 'ge'",
     "output_type": "error",
     "traceback": [
      "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[1;31mAttributeError\u001b[0m                            Traceback (most recent call last)",
      "\u001b[1;32m<ipython-input-1-8781bb79879e>\u001b[0m in \u001b[0;36m<module>\u001b[1;34m()\u001b[0m\n\u001b[0;32m     17\u001b[0m         \u001b[1;31m#  Send reply back to client\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m     18\u001b[0m         \u001b[0msocket\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msend\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34mb\"World\"\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m---> 19\u001b[1;33m \u001b[0masyncio\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mge\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m",
      "\u001b[1;31mAttributeError\u001b[0m: module 'asyncio' has no attribute 'ge'"
     ]
    }
   ],
   "source": [
    "import time\n",
    "import zmq\n",
    "import asyncio\n",
    "context = zmq.Context()\n",
    "socket = context.socket(zmq.REP)\n",
    "socket.bind(\"tcp://*:5555\")\n",
    "\n",
    "async def server():\n",
    "    while True:\n",
    "        #  Wait for next request from client\n",
    "        message = socket.recv()\n",
    "        print(\"Received request: %s\" % message)\n",
    "\n",
    "        #  Do some 'work'\n",
    "        time.sleep(1)\n",
    "\n",
    "        #  Send reply back to client\n",
    "        socket.send(b\"World\")\n",
    "loop=asyncio.get_event_loop()\n",
    "loop.creat_task(server())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "import threading\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "def sayHello():\n",
    "    print(\"hello\")\n",
    "def say(call=\"hi\"):\n",
    "    print(call)\n",
    "    \n",
    "for i in range(3):\n",
    "    t=threading.Timer(3,sayHello)\n",
    "    t.start()\n",
    "    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "say\n"
     ]
    }
   ],
   "source": [
    "import time, threading\n",
    "\n",
    "# 新线程执行的代码:\n",
    "def loop():\n",
    "   print(\"2\")\n",
    "\n",
    "\n",
    "t = threading.Thread (target=say,args=(\"say\",))\n",
    "t.start()\n",
    "t.join()\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 线程池 #https://www.jianshu.com/p/b9b3d66aa0be"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "import time,os\n",
    "from concurrent.futures import ThreadPoolExecutor,as_completed\n",
    "def sayHello(say=\"hello\",who=\"name\"):\n",
    "    print(time.time())\n",
    "    time.sleep(5)\n",
    "    print(say,who)\n",
    "    print('Run child process %s '% os.getpid())\n",
    "    return 1\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1536574434.0684638\n",
      "1536574434.0704687\n",
      "end 1536574434.0734642\n",
      "('hi', 'tom') name\n",
      "Run child process 13684 \n",
      "('hi2', 'tony') name\n",
      "Run child process 13684 \n"
     ]
    }
   ],
   "source": [
    "# map 提交\n",
    "\n",
    "threadPool=ThreadPoolExecutor(3)\n",
    "task=threadPool.map(sayHello,[(\"hi\",\"tom\"),(\"hi2\",\"tony\")]) # 多个函数不同参数 #这样不行\n",
    "\n",
    "print (\"end\",time.time())\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    " # 谁先返回读取谁"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1536567092.8951552\n",
      "1536567092.8981526\n",
      "hi name\n",
      "Run child process 13684 \n",
      "hi2 name\n",
      "Run child process 13684 \n"
     ]
    },
    {
     "ename": "AttributeError",
     "evalue": "'int' object has no attribute 'done'",
     "output_type": "error",
     "traceback": [
      "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[1;31mAttributeError\u001b[0m                            Traceback (most recent call last)",
      "\u001b[1;32m<ipython-input-2-25184aa493ac>\u001b[0m in \u001b[0;36m<module>\u001b[1;34m()\u001b[0m\n\u001b[0;32m      2\u001b[0m \u001b[1;31m#task= [threadPool.submit(sayHello) for i in range(4)]#\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m      3\u001b[0m \u001b[0mtask\u001b[0m\u001b[1;33m=\u001b[0m\u001b[0mthreadPool\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mmap\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0msayHello\u001b[0m\u001b[1;33m,\u001b[0m\u001b[1;33m[\u001b[0m\u001b[1;34m\"hi\"\u001b[0m\u001b[1;33m,\u001b[0m\u001b[1;34m\"hi2\"\u001b[0m\u001b[1;33m]\u001b[0m\u001b[1;33m)\u001b[0m \u001b[1;31m# 多个函数不同参数\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m----> 4\u001b[1;33m \u001b[1;32mwhile\u001b[0m  \u001b[1;32mTrue\u001b[0m \u001b[1;32mnot\u001b[0m \u001b[1;32min\u001b[0m \u001b[1;33m[\u001b[0m\u001b[0mi\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mdone\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m \u001b[1;32mfor\u001b[0m \u001b[0mi\u001b[0m \u001b[1;32min\u001b[0m \u001b[0mtask\u001b[0m\u001b[1;33m]\u001b[0m \u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m      5\u001b[0m     \u001b[1;32mpass\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m      6\u001b[0m \u001b[0mthreadPool\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mshutdown\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mwait\u001b[0m\u001b[1;33m=\u001b[0m\u001b[1;32mFalse\u001b[0m\u001b[1;33m)\u001b[0m \u001b[1;31m#之后无法操作池子\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32m<ipython-input-2-25184aa493ac>\u001b[0m in \u001b[0;36m<listcomp>\u001b[1;34m(.0)\u001b[0m\n\u001b[0;32m      2\u001b[0m \u001b[1;31m#task= [threadPool.submit(sayHello) for i in range(4)]#\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m      3\u001b[0m \u001b[0mtask\u001b[0m\u001b[1;33m=\u001b[0m\u001b[0mthreadPool\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mmap\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0msayHello\u001b[0m\u001b[1;33m,\u001b[0m\u001b[1;33m[\u001b[0m\u001b[1;34m\"hi\"\u001b[0m\u001b[1;33m,\u001b[0m\u001b[1;34m\"hi2\"\u001b[0m\u001b[1;33m]\u001b[0m\u001b[1;33m)\u001b[0m \u001b[1;31m# 多个函数不同参数\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m----> 4\u001b[1;33m \u001b[1;32mwhile\u001b[0m  \u001b[1;32mTrue\u001b[0m \u001b[1;32mnot\u001b[0m \u001b[1;32min\u001b[0m \u001b[1;33m[\u001b[0m\u001b[0mi\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mdone\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m \u001b[1;32mfor\u001b[0m \u001b[0mi\u001b[0m \u001b[1;32min\u001b[0m \u001b[0mtask\u001b[0m\u001b[1;33m]\u001b[0m \u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m      5\u001b[0m     \u001b[1;32mpass\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m      6\u001b[0m \u001b[0mthreadPool\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mshutdown\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mwait\u001b[0m\u001b[1;33m=\u001b[0m\u001b[1;32mFalse\u001b[0m\u001b[1;33m)\u001b[0m \u001b[1;31m#之后无法操作池子\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;31mAttributeError\u001b[0m: 'int' object has no attribute 'done'"
     ]
    }
   ],
   "source": [
    "threadPool=ThreadPoolExecutor(3)\n",
    "task= [threadPool.submit(sayHello) for i in range(4)]#\n",
    "#task=threadPool.map(sayHello,[\"hi\",\"hi2\"]) # 多个函数不同参数\n",
    "while  True not in [i.done() for i in task] :\n",
    "    pass\n",
    "threadPool.shutdown(wait=False) #之后无法操作池子\n",
    "print (\"end\",time.time())\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1536035439.898266\n",
      "hello\n",
      "1536035444.9015968\n",
      "in main: get \n",
      "end 1536035444.9025888\n",
      "hello\n",
      "1536035449.9025638\n",
      "in main: get \n",
      "end 1536035449.9025638\n",
      "hello\n",
      "in main: get \n",
      "end 1536035454.903425\n"
     ]
    }
   ],
   "source": [
    "threadPool=ThreadPoolExecutor(1)\n",
    "task= [threadPool.submit(sayHello) for i in range(3)]#\n",
    "for future in  as_completed(task): # as_completed 是 一个可遍历的，线程结束后 task 任然有结果保存这\n",
    "    data = future.result()#阻塞的 \n",
    "    print(\"in main: get \".format(data))\n",
    "    #threadPool.shutdown(wait=False)\n",
    "    \n",
    "    print (\"end\",time.time())\n",
    "else:\n",
    "    task=[]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[]"
      ]
     },
     "execution_count": 20,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "for future in  as_completed(task):\n",
    "    data = future.result()#阻塞的 \n",
    "    print(\"in main: get \".format(data))\n",
    "    #threadPool.shutdown(wait=False)\n",
    "    print (\"end\",time.time())\n",
    "task"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": 73,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1535282352.1495588\n",
      "1535282352.151671\n",
      "1535282352.1545691\n",
      "hello\n",
      "1535282357.15227\n",
      "end 1535282357.1541631\n",
      "hello\n",
      "hello\n",
      "hello\n"
     ]
    }
   ],
   "source": [
    "from concurrent.futures import wait ,FIRST_COMPLETED\n",
    "threadPool=ThreadPoolExecutor(3)\n",
    "task= [threadPool.submit(sayHello) for i in range(10)]#\n",
    "\n",
    "wait(task,return_when=FIRST_COMPLETED)\n",
    "for i in task:\n",
    "    i.cancel()\n",
    "print (\"end\",time.time())\n",
    "#help(wait)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1535975708.3186097\n",
      "hello\n",
      "1535975713.3276505\n"
     ]
    },
    {
     "ename": "KeyboardInterrupt",
     "evalue": "",
     "output_type": "error",
     "traceback": [
      "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[1;31mKeyboardInterrupt\u001b[0m                         Traceback (most recent call last)",
      "\u001b[1;32m<ipython-input-12-82e02afb0278>\u001b[0m in \u001b[0;36m<module>\u001b[1;34m()\u001b[0m\n\u001b[0;32m      2\u001b[0m \u001b[0mtask\u001b[0m\u001b[1;33m=\u001b[0m \u001b[1;33m[\u001b[0m\u001b[0mthreadPool\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msubmit\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0msayHello\u001b[0m\u001b[1;33m)\u001b[0m \u001b[1;32mfor\u001b[0m \u001b[0mi\u001b[0m \u001b[1;32min\u001b[0m \u001b[0mrange\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;36m3\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m]\u001b[0m\u001b[1;31m#\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m      3\u001b[0m \u001b[1;32mwhile\u001b[0m \u001b[1;36m1\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m----> 4\u001b[1;33m     \u001b[1;34m\"\"\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m",
      "\u001b[1;31mKeyboardInterrupt\u001b[0m: "
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "hello\n",
      "1535975718.3283072\n",
      "hello\n"
     ]
    }
   ],
   "source": [
    "threadPool=ThreadPoolExecutor(1)\n",
    "task= [threadPool.submit(sayHello) for i in range(3)]#\n",
    "while 1:\n",
    "    \"\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 多进程"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "ename": "ModuleNotFoundError",
     "evalue": "No module named 'tmp'",
     "output_type": "error",
     "traceback": [
      "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[1;31mModuleNotFoundError\u001b[0m                       Traceback (most recent call last)",
      "\u001b[1;32m<ipython-input-1-8743b679f9f4>\u001b[0m in \u001b[0;36m<module>\u001b[1;34m()\u001b[0m\n\u001b[0;32m      2\u001b[0m \u001b[1;32mimport\u001b[0m \u001b[0mos\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mtime\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mrandom\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m      3\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m----> 4\u001b[1;33m \u001b[1;32mfrom\u001b[0m \u001b[0mtmp\u001b[0m \u001b[1;32mimport\u001b[0m \u001b[0mwrite\u001b[0m\u001b[1;33m,\u001b[0m\u001b[0mread\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m      5\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m      6\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;31mModuleNotFoundError\u001b[0m: No module named 'tmp'"
     ]
    }
   ],
   "source": [
    "from multiprocessing import Process, Queue\n",
    "import os, time, random\n",
    "\n",
    "from tmp import write,read\n",
    "\n",
    "\n",
    "print(os.getpid())\n",
    "# 父进程创建Queue，并传给各个子进程：\n",
    "q = Queue()\n",
    "pw = Process(target=write, args=(q,))\n",
    "pr = Process(target=read, args=(q,))\n",
    "# 启动子进程pw，写入:\n",
    "pw.start()\n",
    "# 启动子进程pr，读取:\n",
    "pr.start()\n",
    "# 等待pw结束:\n",
    "pw.join()\n",
    "# pr进程里是死循环，无法等待其结束，只能强行终止:\n",
    "pr.terminate()\n",
    "#  在后台能看到shuchu "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 进程池"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "ename": "ModuleNotFoundError",
     "evalue": "No module named 'tmp'",
     "output_type": "error",
     "traceback": [
      "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[1;31mModuleNotFoundError\u001b[0m                       Traceback (most recent call last)",
      "\u001b[1;32m<ipython-input-2-4c840c466ea7>\u001b[0m in \u001b[0;36m<module>\u001b[1;34m()\u001b[0m\n\u001b[0;32m      1\u001b[0m \u001b[1;32mfrom\u001b[0m \u001b[0mmultiprocessing\u001b[0m \u001b[1;32mimport\u001b[0m \u001b[0mPool\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m----> 2\u001b[1;33m \u001b[1;32mfrom\u001b[0m \u001b[0mtmp\u001b[0m \u001b[1;32mimport\u001b[0m \u001b[0msayHello\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m      3\u001b[0m \u001b[1;32mimport\u001b[0m \u001b[0mtime\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m      4\u001b[0m \u001b[1;32mwith\u001b[0m \u001b[0mPool\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;36m3\u001b[0m\u001b[1;33m)\u001b[0m \u001b[1;32mas\u001b[0m \u001b[0mpl\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m      5\u001b[0m     \u001b[0mres\u001b[0m\u001b[1;33m=\u001b[0m\u001b[0mpl\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mapply_async\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0msayHello\u001b[0m\u001b[1;33m,\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m\"hi\"\u001b[0m\u001b[1;33m,\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m)\u001b[0m \u001b[1;31m#非阻塞的\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;31mModuleNotFoundError\u001b[0m: No module named 'tmp'"
     ]
    }
   ],
   "source": [
    "from multiprocessing import Pool \n",
    "from tmp import sayHello\n",
    "import time \n",
    "with Pool(3) as pl:\n",
    "    res=pl.apply_async(sayHello,(\"hi\",)) #非阻塞的 \n",
    "    pl.join()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "with Pool(3) as pl:\n",
    "    pl.apply(sayHello,(\"hi\",)) #参数必须是元组 不然不运行进程"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "with Pool(3) as pl:\n",
    "    pl.map(sayHello,(\"hi\",)) #参数必须是元组 不然不运行进程"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Queue"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "False\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "[124, 35, 4]"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "p\n",
      "p\n",
      "p\n",
      "p\n"
     ]
    }
   ],
   "source": [
    "import queue,time,threading\n",
    "q=queue.Queue(3)\n",
    "q.put([124,35,4]) # full 是阻塞\n",
    "def product( ):\n",
    "    for i in range(20):\n",
    "        time.sleep(1)\n",
    "        print(\"p\")\n",
    "        q.put([2300,34,23,4])\n",
    "threading.Thread(target=product).start()\n",
    "print(q.empty())\n",
    "q.get()# 空时会阻塞"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "q.get()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 66,
   "metadata": {},
   "outputs": [],
   "source": [
    "try:\n",
    "    print(q.get(block=False) )#  为空时抛出异常\n",
    "except queue.Empty:\n",
    "    pass"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# python3 异步\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "import asyncio ,time\n",
    "\n",
    "async def hello_world():\n",
    "    await  asyncio.sleep(1) #一个协程里可以启动另外一个协程，并等待它完成返回结果\n",
    "    #await关键字添加了一个新的协程到循环里，而不需要明确地添加协程到这个事件循环里。\n",
    "    print(\"Hello World!\")\n",
    "\n",
    "loop = asyncio.get_event_loop() # 开始事件循环\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<Task pending coro=<hello_world() running at <ipython-input-14-48f0c34c12bc>:3>>"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Hello World!\n"
     ]
    }
   ],
   "source": [
    "asyncio.ensure_future(hello_world()) "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "ename": "RuntimeError",
     "evalue": "This event loop is already running",
     "output_type": "error",
     "traceback": [
      "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[1;31mRuntimeError\u001b[0m                              Traceback (most recent call last)",
      "\u001b[1;32m<ipython-input-16-81232acc2d83>\u001b[0m in \u001b[0;36m<module>\u001b[1;34m()\u001b[0m\n\u001b[1;32m----> 1\u001b[1;33m \u001b[0mloop\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mrun_until_complete\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mhello_world\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m",
      "\u001b[1;32mc:\\program files\\python36\\lib\\asyncio\\base_events.py\u001b[0m in \u001b[0;36mrun_until_complete\u001b[1;34m(self, future)\u001b[0m\n\u001b[0;32m    452\u001b[0m         \u001b[0mfuture\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0madd_done_callback\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0m_run_until_complete_cb\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    453\u001b[0m         \u001b[1;32mtry\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m--> 454\u001b[1;33m             \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mrun_forever\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m    455\u001b[0m         \u001b[1;32mexcept\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    456\u001b[0m             \u001b[1;32mif\u001b[0m \u001b[0mnew_task\u001b[0m \u001b[1;32mand\u001b[0m \u001b[0mfuture\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mdone\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m \u001b[1;32mand\u001b[0m \u001b[1;32mnot\u001b[0m \u001b[0mfuture\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mcancelled\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32mc:\\program files\\python36\\lib\\asyncio\\base_events.py\u001b[0m in \u001b[0;36mrun_forever\u001b[1;34m(self)\u001b[0m\n\u001b[0;32m    406\u001b[0m         \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_check_closed\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    407\u001b[0m         \u001b[1;32mif\u001b[0m \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mis_running\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m--> 408\u001b[1;33m             \u001b[1;32mraise\u001b[0m \u001b[0mRuntimeError\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m'This event loop is already running'\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m    409\u001b[0m         \u001b[1;32mif\u001b[0m \u001b[0mevents\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_get_running_loop\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m \u001b[1;32mis\u001b[0m \u001b[1;32mnot\u001b[0m \u001b[1;32mNone\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    410\u001b[0m             raise RuntimeError(\n",
      "\u001b[1;31mRuntimeError\u001b[0m: This event loop is already running"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Hello World!\n"
     ]
    }
   ],
   "source": [
    "loop.run_until_complete(hello_world())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 在jupyter 中运行 asyncio\n",
    "\n",
    "因此，对于在Jupyter Notebook上运行的任何asyncio功能，您无法调用a run_until_complete()，因为您将从中接收的循环asyncio.get_event_loop()将处于活动状态。\n",
    "\n",
    "相反，您必须将任务添加到当前循环：\n",
    "\n",
    "import asyncio\n",
    "loop = asyncio.get_event_loop()\n",
    "loop.create_task(some_async_function())\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<generator object Queue.get at 0x0000023DB6B2C3B8>\n",
      "<generator object Queue.get at 0x0000023DB6B2C3B8>\n",
      "<generator object Queue.get at 0x0000023DB6B2C3B8>\n",
      "<generator object Queue.get at 0x0000023DB6B2C3B8>\n",
      "<generator object Queue.get at 0x0000023DB6B2C3B8>\n",
      "<generator object Queue.get at 0x0000023DB6B2C3B8>\n",
      "<generator object Queue.get at 0x0000023DB6B2C3B8>\n",
      "<generator object Queue.get at 0x0000023DB6B2C3B8>\n",
      "<generator object Queue.get at 0x0000023DB6B2C3B8>\n",
      "<generator object Queue.get at 0x0000023DB6B2C3B8>\n",
      "<generator object Queue.get at 0x0000023DB6BF56D0>\n",
      "<generator object Queue.get at 0x0000023DB6BF56D0>\n",
      "<generator object Queue.get at 0x0000023DB6BF56D0>\n",
      "<generator object Queue.get at 0x0000023DB6BF56D0>\n",
      "<generator object Queue.get at 0x0000023DB6BF56D0>\n",
      "<generator object Queue.get at 0x0000023DB6BF56D0>\n",
      "<generator object Queue.get at 0x0000023DB6BF56D0>\n",
      "<generator object Queue.get at 0x0000023DB6BF56D0>\n",
      "<generator object Queue.get at 0x0000023DB6BF56D0>\n",
      "<generator object Queue.get at 0x0000023DB6BF56D0>\n",
      "<generator object Queue.get at 0x0000023DB6BF54C0>\n",
      "<generator object Queue.get at 0x0000023DB6BF54C0>\n",
      "<generator object Queue.get at 0x0000023DB6BF54C0>\n",
      "<generator object Queue.get at 0x0000023DB6BF54C0>\n",
      "<generator object Queue.get at 0x0000023DB6BF54C0>\n",
      "<generator object Queue.get at 0x0000023DB6BF54C0>\n",
      "<generator object Queue.get at 0x0000023DB6BF54C0>\n",
      "<generator object Queue.get at 0x0000023DB6BF54C0>\n",
      "<generator object Queue.get at 0x0000023DB6BF54C0>\n",
      "<generator object Queue.get at 0x0000023DB6BF54C0>\n"
     ]
    }
   ],
   "source": [
    "# await \n",
    "import asyncio.queues\n",
    "import time\n",
    "async def putInCh(ch):\n",
    "    \n",
    "    for i in range(3):\n",
    "        time.sleep(1)\n",
    "        ch.put(20)\n",
    "\n",
    "async def test():\n",
    "    ch=asyncio.Queue(5)\n",
    "    await putInCh(ch)\n",
    "    for i in range(10):\n",
    "        print( ch.get())\n",
    "        \n",
    "if __name__ == '__main__':\n",
    "    loop = asyncio.get_event_loop() # 开始事件循环\n",
    "\n",
    "    \n",
    "    loop.create_task(test())\n",
    "    loop.create_task(test())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "code_folding": [
     4
    ]
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "123\n",
      "WWWWWW\n"
     ]
    }
   ],
   "source": [
    "import asyncio\n",
    "import time\n",
    "async def showWdelay():\n",
    "    time.sleep(1)\n",
    "    print(\"WWWWWW\")\n",
    "async def test():\n",
    "    await showWdelay()\n",
    "        \n",
    "if __name__ == '__main__':\n",
    "    loop = asyncio.get_event_loop() # 开始事件循环\n",
    "\n",
    "    \n",
    "    loop.create_task(test())\n",
    "    print(123)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# yield 与事件循环？\n",
    "yield 是不能在async 中使用的，那么为什么？ 还可以迭代吗？"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "123\n",
      "WWWWWW\n",
      "0\n"
     ]
    }
   ],
   "source": [
    "import asyncio\n",
    "import time\n",
    "async def showWdelay():\n",
    "    time.sleep(1)\n",
    "    print(\"WWWWWW\")\n",
    "    for i in range(10):\n",
    "        return i\n",
    "        #yield\n",
    "async def test():\n",
    "    print(await showWdelay())\n",
    "        \n",
    "if __name__ == '__main__':\n",
    "    loop = asyncio.get_event_loop() # 开始事件循环\n",
    "\n",
    "    \n",
    "    loop.create_task(test())\n",
    "    print(123)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 异步迭代器\n",
    "[async with ；async for ](https://blog.csdn.net/qq_35304570/article/details/78209612)\n",
    "\n",
    "**异步迭代对象必须实现 __aiter__ 等**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "import asyncio\n",
    "import time\n",
    "class AsyncIteratorWrapper:\n",
    "    def __init__(self, obj):\n",
    "        self._it = iter(obj)\n",
    " \n",
    "    def __aiter__(self):\n",
    "        return self\n",
    " \n",
    "    async def __anext__(self):\n",
    "        try:\n",
    "            value = next(self._it)\n",
    "        except StopIteration:\n",
    "            raise StopAsyncIteration\n",
    "        return value\n",
    "\n",
    "def showWdelay(iterator):\n",
    "    time.sleep(1)\n",
    "    print(\"WWWWWW\")\n",
    "    return AsyncIteratorWrapper(iterator)\n",
    "#     async for i in AsyncIteratorWrapper(iterator):\n",
    "#         print(i)\n",
    "#         print(\"A\")\n",
    "##         return i\n",
    "        \n",
    "async def test():\n",
    "    a=( (i,time.sleep(1)) for i in range(10))\n",
    "    aa=showWdelay(a).__aiter__()\n",
    "    print( await aa.__anext__())\n",
    "    print( \"如果这是异步的哪么就应该同时打印出来\")\n",
    "    print( await aa.__anext__())\n",
    "    \n",
    "#     print(await showWdelay(a))\n",
    "#     print(\"插入一条\")\n",
    "#     print(await showWdelay(a))\n",
    "#     print(await showWdelay(a))\n",
    "async def test2():\n",
    "    a=( (i,time.sleep(1)) for i in range(3))\n",
    "    aa=showWdelay(a).__aiter__()\n",
    "    async for i in aa:\n",
    "        print(i)\n",
    "        yield i\n",
    "    print(12334)\n",
    "\n",
    "loop = asyncio.get_event_loop() # 开始事件循环\n",
    "\n",
    "    \n",
    "#     loop.create_task(test())\n",
    "#     print(123)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "123\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Exception in default exception handler\n",
      "Traceback (most recent call last):\n",
      "  File \"c:\\program files\\python36\\lib\\asyncio\\base_events.py\", line 1293, in call_exception_handler\n",
      "    self.default_exception_handler(context)\n",
      "  File \"c:\\program files\\python36\\lib\\asyncio\\base_events.py\", line 1265, in default_exception_handler\n",
      "    value = repr(value)\n",
      "  File \"c:\\program files\\python36\\lib\\asyncio\\base_tasks.py\", line 15, in _task_repr_info\n",
      "    coro = coroutines._format_coroutine(task._coro)\n",
      "  File \"c:\\program files\\python36\\lib\\asyncio\\coroutines.py\", line 276, in _format_coroutine\n",
      "    assert iscoroutine(coro)\n",
      "AssertionError\n"
     ]
    }
   ],
   "source": [
    "loop.create_task(test2())\n",
    "print(123)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 62,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "123\n",
      "WWWWWW\n",
      "0\n"
     ]
    }
   ],
   "source": [
    "loop.create_task(test())\n",
    "print(123)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "2\n"
     ]
    }
   ],
   "source": [
    "def d(a):\n",
    "    print(a[2])\n",
    "a={1:[2]}\n",
    "for i in a.values():\n",
    "    for v in i:\n",
    "        print(v)\n"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.6"
  },
  "toc": {
   "base_numbering": 1,
   "nav_menu": {},
   "number_sections": true,
   "sideBar": true,
   "skip_h1_title": false,
   "title_cell": "Table of Contents",
   "title_sidebar": "Contents",
   "toc_cell": false,
   "toc_position": {},
   "toc_section_display": true,
   "toc_window_display": true
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
